package com.xinxin.socket.business;

import com.xinxin.base.util.JsonUtil;
import com.xinxin.socket.constant.CommandId;
import com.xinxin.socket.context.ChannelLoginContext;
import com.xinxin.socket.msg.NetMsg;
import com.xinxin.socket.sample.context.TopicContext;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.util.Optional;

/**
 * @author ys
 * @date 2020/8/11
 */
@Slf4j
public class ServerMessageSender {

    private static final ChannelLoginContext CHANNEL_LOGIN_CONTEXT = ChannelLoginContext.getInstance();
    private static final TopicContext TOPIC_CONTEXT = TopicContext.getInstance();

    private static NetMsg buildNetMsg(int type, byte[] body) {
        return new NetMsg().setCmdId(type).body(body);
    }

    public static void pushMessage(String topicName, Object message) {
        log.info("Server push message");
        Optional.ofNullable(TOPIC_CONTEXT.getTopicQueue(topicName)).ifPresent(topicQueue -> {
            try {
                log.info("Send to topic: {}, client size: {}", topicName, topicQueue.size());
                NetMsg netMsg = buildNetMsg(CommandId.SRV_PUSH, JsonUtil.writeValueAsBytes(message));
                topicQueue.forEach(ch -> ch.writeAndFlush(netMsg)
                        .addListeners((GenericFutureListener) future -> {
                            if (future.isSuccess()) {
                                log.info("Send success: {}", CHANNEL_LOGIN_CONTEXT.getLoginId(ch));
                            }
                        }, ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE));
            } catch (Exception e) {
                log.error("server push error", e);
            }
        });
    }

    public static boolean pushMessage(Long userId, int commandId, Object message) {
        try {
            String msg = JsonUtil.writeValueAsString(message);
            log.info("Server push message, userId: {}, commandId: {}, message: {}", userId, commandId, msg);

            ChannelFuture channelFuture = CHANNEL_LOGIN_CONTEXT.pushMessage(userId, commandId, msg);
            if (channelFuture == null) {
                log.warn("User not online.");
                return false;
            }

            return channelFuture.sync().isSuccess();
        } catch (Exception e) {
            log.error("server push error", e);
            return false;
        }
    }
}
